use hashbrown::HashMap;
use honeybee::prelude::*;
use honeybee::Result;

use crate::wait_online_user::{KickoutOldUserSesscss, ResetStream, WaitOnlineUserActor};

struct UserSession {
    session_key: String,
    sequence_id: u128,
    app_node: String,
    stream: Addr<TcpStreamActor>,
}

impl UserSession {
    fn new(
        session_key: String,
        sequence_id: u128,
        app_node: String,
        stream: Addr<TcpStreamActor>,
    ) -> Self {
        Self {
            session_key,
            sequence_id,
            app_node,
            stream,
        }
    }
}

struct WaitOnlineUserSession {
    app_node: String,
    addr: Addr<WaitOnlineUserActor>,
}

pub struct Users {
    self_addr: Option<Addr<Self>>,
    users: HashMap<i64, UserSession>,
    wait_online_users: HashMap<String, WaitOnlineUserSession>,
}

impl Users {
    pub async fn new() -> Self {
        Self {
            self_addr: None,
            users: HashMap::new(),
            wait_online_users: HashMap::new(),
        }
    }
}

#[async_trait::async_trait]
impl Actor for Users {
    fn name(&self) -> &str {
        "Users"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        info!("{} started", self.name());
        let addr = ctx.address();
        self.self_addr = Some(addr.clone());
        ctx.register_center(self.name()).await?;

        let router = ctx.get_router().await?;
        router.register_rpc::<proto::UscUserRegister>();
        router.register_rpc::<proto::UscUserUnregister>();
        router.register::<proto::UserMessageForward>();
        router.register::<proto::UserMessageApp>();

        //ctx.send_later(StopStream, Duration::from_secs(10));
        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        if let Err(err) = ctx.unregister_center().await {
            error!("users unregister_center failed:{}", err);
        }
        info!("{} stoped", self.name());
    }
}

//test
#[message]
struct StopStream;
#[async_trait::async_trait]
impl Handler<StopStream> for Users {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _: StopStream) -> Result<()> {
        info!("xxxxx stop stream");
        for user in self.users.values() {
            user.stream.stop(None).ok();
        }
        Ok(())
    }
}

#[message]
pub struct AppConnected(pub String, pub Addr<TcpStreamActor>);
#[async_trait::async_trait]
impl Handler<AppConnected> for Users {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: AppConnected) -> Result<()> {
        for user in self.users.values_mut() {
            if user.app_node == msg.0 {
                user.stream = msg.1.clone();
            }
        }
        for wait_online_user in self.wait_online_users.values() {
            if wait_online_user.app_node == msg.0 {
                let _ = wait_online_user.addr.send(ResetStream(msg.1.clone()));
            }
        }
        Ok(())
    }
}

#[message]
pub struct RemoveWaitOnlineUser(pub String);
#[async_trait::async_trait]
impl Handler<RemoveWaitOnlineUser> for Users {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: RemoveWaitOnlineUser) -> Result<()> {
        self.wait_online_users.remove(&msg.0);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetRpcMsg<proto::UscUserRegister>> for Users {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetRpcMsg<proto::UscUserRegister>,
    ) -> Result<()> {
        let (uid, session, source, msg) = (msg.msg.uid, msg.session, msg.source, msg.msg);
        if let Some(user) = self.users.get(&uid) {
            if user.app_node == msg.app_node && user.sequence_id > msg.sequence_id {
                return Err(anyhow!("online failed with the earlier sequence_id"));
            }
        }
        let old_user = self.users.insert(
            uid,
            UserSession::new(
                msg.session_key,
                msg.sequence_id,
                msg.app_node.clone(),
                source.clone(),
            ),
        );
        //如果有老玩家在线，让新玩家去新的actor上去等,等老玩家下线反馈成功后再通知新玩家上线
        if let Some(old_user) = old_user {
            let addr = WaitOnlineUserActor::new(
                self.self_addr.as_ref().unwrap().clone(),
                old_user.session_key.clone(),
                uid,
                session,
                source,
            )
            .start()
            .await?;
            let wait_online_user_session = WaitOnlineUserSession {
                app_node: msg.app_node,
                addr,
            };
            //如果有正在等待上线的玩家会通知ta上线失败,让最后一个上线玩家等待最早的玩家下线成功
            self.wait_online_users
                .insert(old_user.session_key.clone(), wait_online_user_session);
            info!("users:{} register, repeated online notify kickout", uid);
            old_user.stream.send_to_app_user(
                uid,
                proto::UscRepeatOnlineKickout {
                    session_key: old_user.session_key,
                },
            );
            return Err(RpcPending.into());
        }
        info!("user:{} registed", uid);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetRpcMsg<proto::UscUserUnregister>> for Users {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetRpcMsg<proto::UscUserUnregister>,
    ) -> Result<()> {
        let msg = msg.msg;
        if let Some(wait_online_user) = self.wait_online_users.remove(&msg.session_key) {
            let _ = wait_online_user.addr.send(KickoutOldUserSesscss);
        }
        if let Some(user) = self.users.get(&msg.uid) {
            if user.session_key != msg.session_key {
                return Ok(());
            }
        }
        if self.users.remove(&msg.uid).is_none() {
            error!("users unregister:{} not found user", msg.uid);
        }
        info!("user:{} unregisted", msg.uid);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::UserMessageForward>> for Users {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::UserMessageForward>,
    ) -> Result<()> {
        let msg = msg.msg;
        if let Some(user) = self.users.get(&msg.uid) {
            user.stream.send_remote(msg)?;
        } else {
            error!(
                "forward msg:{} to user:{} failed, not found user",
                msg.msg_id, msg.uid
            );
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<NetMsg<proto::UserMessageApp>> for Users {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: NetMsg<proto::UserMessageApp>,
    ) -> Result<()> {
        let msg = msg.msg;
        if let Some(user) = self.users.get(&msg.uid) {
            user.stream.send_remote(msg)?;
        } else {
            error!(
                "msg:{} to app user:{} failed, not found user",
                msg.msg_id, msg.uid
            );
        }
        Ok(())
    }
}

#[message]
pub struct UmsgNotify {
    pub uid: i64,
    pub object_id: String,
    pub msg_type: umsg_db::UmsgType,
    pub msg_bytes: Vec<u8>,
}
#[async_trait::async_trait]
impl Handler<UmsgNotify> for Users {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: UmsgNotify) -> Result<()> {
        if let Some(user) = self.users.get(&msg.uid) {
            user.stream.send_to_app_user(
                msg.uid,
                proto::NewUmsgNotify {
                    object_id: msg.object_id,
                    msg_type: msg.msg_type,
                    msg_bytes: msg.msg_bytes,
                },
            );
        }
        Ok(())
    }
}
